package com.jch.watermaker;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import java.time.Duration;
import java.util.Random;
import java.util.UUID;

public class WaterMarkTest {

    public static void main(String[] args) throws Exception{

        //创建环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //模拟数据
        DataStreamSource<Order> source = env.addSource(new SourceFunction<Order>() {

            private boolean flag=true;
            @Override
            public void run(SourceContext<Order> sourceContext) throws Exception {

                Random random = new Random();

                while (flag){
                    String orderid = UUID.randomUUID().toString();
                    int userid = random.nextInt(4);
                    int money = random.nextInt(100);
                    long eventtime = System.currentTimeMillis() - random.nextInt(5) * 1000;
                    sourceContext.collect(new Order(orderid,userid,money,eventtime));
                    Thread.sleep(1000);
                }
            }

            @Override
            public void cancel() {
                flag=false;
            }
        });

        //添加水位线

         SingleOutputStreamOperator<Order> orderSingleOutputStreamOperator
                 =source.assignTimestampsAndWatermarks(WatermarkStrategy.<Order>
                 forBoundedOutOfOrderness(Duration.ofSeconds(3))
                 .withTimestampAssigner((event, timestamp) -> event.getEventTime()));

         //秋季呢
        SingleOutputStreamOperator<Order> sum = orderSingleOutputStreamOperator
                .keyBy(Order::getUserId)
                .window(TumblingProcessingTimeWindows.of(Time.seconds(5)))
                .sum("money");

        //打印
        sum.print();
        //启动
        env.execute();
    }


    @Data
    @NoArgsConstructor
    @AllArgsConstructor
    public static class Order{

        private String orderId;
        private Integer userId;
        private Integer money;
        private Long eventTime;

    }



}
